﻿//
// Created by 25007 on 2023/2/20.
//

#ifndef FFMPEG_PLAYER_ACL_MQTT_SERVER_ADAPTER_H
#define FFMPEG_PLAYER_ACL_MQTT_SERVER_ADAPTER_H

#include <iostream>
#include <iomanip>
#include <map>
# if acl
#include <mqtt_client_cpp.hpp>
#include <utility>
#include "net/base_net.h"
#include "log.h"


#include "acl_cpp/lib_acl.hpp"
#include "fiber/fiber.hpp"

class mqtt_async_client: public BaseNet{

    /// @brief 客户端id
    std::string m_client_id_;

    /// @brief 是否清理session
    bool clean_session_;

    /// @brief 连接
    std::function<void()> connect;

    /// @brief 重新连接
    std::function<void()> reconnect;
public:
    mqtt_async_client(std::string mClientId, bool cleanSession);

public:
    void run() override;
};

class mqtt_client;

class mqtt_clients {
public:
    mqtt_clients(const char* topic) : topic_(topic) {}
    ~mqtt_clients(void) {}

    const char* get_topic(void) const {
        return topic_.c_str();
    }

    std::set<mqtt_client*>& get_clients(void) {
        return clients_;
    }

    bool add(mqtt_client* client);
    bool del(mqtt_client* client);
    bool empty(void) const {
        return clients_.empty();
    }

    void clear_all(void);

private:
    acl::string topic_;
    std::set<mqtt_client*> clients_;
};

class mqtt_manager {
public:
    mqtt_manager(void) {}
    ~mqtt_manager(void);

    bool add(const char* topic, mqtt_client* client);
    bool del(mqtt_client* client);

    std::set<mqtt_client*>* get_clients(const char* topic) {
        std::map<acl::string, mqtt_clients*>::iterator it =
                manager_.find(topic);
        if (it == manager_.end()) {
            return NULL;
        }
        return &it->second->get_clients();
    }

private:
    std::map<acl::string, mqtt_clients*> manager_;

    void del(const char* topic, mqtt_client* client);
};

class mqtt_client : public acl::mqtt_aclient {
public:
    mqtt_client(acl::aio_handle& handle, mqtt_manager& manager);

    const std::set<acl::string>& get_topics() const {
        return topics_;
    }

protected:
    ~mqtt_client() override;

    // @override
    void destroy() override { delete this; }

    // @override
    bool on_open() override;

    // @override
    void on_disconnect() override {
        printf("client disconnect\r\n");
        manager_.del(this);
    }

    // @override
    bool on_header(const acl::mqtt_header& header) override;

    // @override
    bool on_body(const acl::mqtt_message& body) override;

private:
    mqtt_manager& manager_;
    std::set<acl::string> topics_;

    bool handle_connect(const acl::mqtt_message& body);
    bool handle_subscribe(const acl::mqtt_message& body);
    bool handle_pingreq(const acl::mqtt_message& body);
    bool handle_pingresp(const acl::mqtt_message& body);
    bool handle_disconnect(const acl::mqtt_message& body);
    bool handle_publish(const acl::mqtt_message& body);
    bool handle_puback(const acl::mqtt_message& body);

    void forward_publish(const acl::mqtt_publish& publish);
    void forward_publish(mqtt_client& client, const acl::mqtt_publish& in);
};


//////////////////////////////////////////////////////////////////////////////

class server_callback : public acl::aio_accept_callback {
public:
    server_callback(acl::aio_handle& handle) : handle_(handle) {}
    ~server_callback(void) {}

protected:
    // @override
    bool accept_callback(acl::aio_socket_stream* conn) {
        printf("connect from %s, fd=%llu\r\n", conn->get_peer(true),
               conn->sock_handle());

        acl::mqtt_aclient* client = new mqtt_client(handle_, manager_);
        if (client->open(conn)) {
            return true;
        }

        printf("open one mqtt client failed\r\n");
        client->destroy();
        conn->close();
        return true;
    }

private:
    acl::aio_handle& handle_;
    mqtt_manager manager_;
};
#endif

#endif //FFMPEG_PLAYER_ACL_MQTT_SERVER_ADAPTER_H
